package com.example.realtimeeb.ods

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.clients.consumer.ConsumerConfig
import java.util.Properties

class SourceKafka {

  def getKafkaSource(topic:String):FlinkKafkaConsumer[String] = {
    val props = new Properties()
    props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux122:9092")
    props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer")
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
    // 便于调试
//    props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
//    props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest")
    new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), props)
  }

}
